package com.desheng.bigdata.flink.table

import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.types.Row
/**
  * 基于flinksql的wordcount的统计
  */
object _04FlinkSQL2WordCountOps {
    def main(args: Array[String]): Unit = {

        //创建流对应的Env
        val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
        //基于流Env构建StreamTableEnv
        val tblEnv = StreamTableEnvironment.create(streamEnv)
        val line = streamEnv.socketTextStream("bigdata01", 9999)

        //转化为window再进行计算

        var tbl = tblEnv.fromDataStream(line).as("line")
        tblEnv.registerTable("tmp", tbl)
        //直接统计wordcount 这种写法是错误的，详见_09FlinkTable2UDTFOps
        tbl = tblEnv.sqlQuery(
            """
              |select
              |  t.word,
              |  count(1) counts
              |from (
              |  select
              |     explode(split(line, '\\s+')) word
              |  from tmp
              |) t
              |group by t.word
            """.stripMargin)
        tblEnv.toAppendStream[Row](tbl).print()
        streamEnv.execute()
    }
}
